use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;

use async_std::net::TcpStream;
use futures::channel::oneshot;
use rand::seq::SliceRandom;
use rand::thread_rng;
use serde::Serialize;
use xactor::{message, spawn, Actor, Addr, Context, Handler, Message};

use crate::actors::tcpstream::TcpStreamActor;
use crate::message::RpcResponseError;
use crate::prelude::*;
use crate::Result;

struct Node {
    addr: String,
    #[allow(dead_code)]
    remote_node_type: String,
    remote_node_name: String,
    stream: Option<Addr<TcpStreamActor>>,
}

impl Node {
    fn new(addr: String) -> Self {
        Self {
            addr,
            remote_node_type: "".into(),
            remote_node_name: "".into(),
            stream: None,
        }
    }
}

impl std::fmt::Display for Node {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{},{}", self.remote_node_type, self.remote_node_name)
    }
}

pub struct TcpConnectersActor {
    actor_name: String,
    node_type: String,
    node_name: String,
    self_addr: Option<Addr<Self>>,
    message_register: Arc<MessageRegister>,
    reconnect_interval: Duration,
    nodes: Vec<Node>,
}

impl TcpConnectersActor {
    pub fn new(
        actor_name: &str,
        node_type: &str,
        node_name: &str,
        remote_addrs: &[String],
        reconnect_interval: Duration,
        message_register: Arc<MessageRegister>,
    ) -> Self {
        let nodes = remote_addrs
            .iter()
            .map(|addr| Node::new(addr.clone()))
            .collect();

        Self {
            self_addr: None,
            actor_name: actor_name.into(),
            node_type: node_type.into(),
            node_name: node_name.into(),
            message_register,
            reconnect_interval,
            nodes,
        }
    }
}

impl std::fmt::Display for TcpConnectersActor {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        write!(f, "{}", self.name())
    }
}

#[async_trait::async_trait]
impl Actor for TcpConnectersActor {
    fn name(&self) -> &str {
        &self.actor_name
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> Result<()> {
        let addr = ctx.address();
        self.self_addr = Some(addr.clone());
        ctx.register_center(&self.actor_name).await?;
        for (index, _) in self.nodes.iter().enumerate() {
            let _ = addr.send(Connect(index));
        }
        Ok(())
    }

    async fn stopped(&mut self, ctx: &mut Context<Self>) {
        if let Err(err) = ctx.unregister_center().await {
            error!("{} unregister_center failed:{}", self, err);
        }
        info!("{} stoped", self);
    }
}

#[message]
struct Connect(usize);
#[async_trait::async_trait]
impl Handler<Connect> for TcpConnectersActor {
    async fn handle(&mut self, ctx: &mut Context<Self>, msg: Connect) -> Result<()> {
        let index = msg.0;
        if self.nodes[index].stream.is_some() {
            return Err(anyhow!("{} already connected", self.nodes[index].addr));
        }

        info!("try connect to {}@{} ..", self, self.nodes[index].addr);

        let remote_addr = self.nodes[index].addr.clone();
        match TcpStream::connect(&remote_addr).await {
            Ok(stream) => {
                stream.set_nodelay(true)?;
                let mut framed_write =
                    FramedWrite::new(stream.clone().compat(), HoneybeeMsgCodec::new());
                let mut framed_read =
                    FramedRead::new(stream.clone().compat(), HoneybeeMsgCodec::new());
                framed_write
                    .send_ext(TcpRegister::new(&self.node_type, &self.node_name))
                    .await?;

                let msg: TcpRegister = framed_read.read_msg().await?;

                let remote_node_type = msg.node_type;
                let remote_node_name = msg.node_name;

                let stream_addr = TcpStreamActor::new(
                    &remote_node_type,
                    &remote_node_name,
                    framed_write,
                    framed_read,
                    self.message_register.clone(),
                    None,
                )
                .start()
                .await
                .unwrap();

                self.nodes[index].stream = Some(stream_addr.clone());
                info!(
                    "{}({}.{}@{}) connected",
                    self, remote_node_type, remote_node_name, remote_addr
                );
                let addr = ctx.address();
                let reconnect_interval = self.reconnect_interval;
                spawn(async move {
                    stream_addr.wait_for_stop().await;
                    sleep(reconnect_interval).await;
                    let _ = addr.send(DisConnected(index));
                });
            }
            Err(err) => {
                error!(
                    "connect to {}@{} failed err:{}, try reconnect after {}s",
                    self,
                    self.nodes[index].addr,
                    err,
                    self.reconnect_interval.as_secs()
                );
                ctx.send_later(Connect(index), self.reconnect_interval);
            }
        }
        Ok(())
    }
}

#[message]
struct DisConnected(usize);
#[async_trait::async_trait]
impl Handler<DisConnected> for TcpConnectersActor {
    async fn handle(&mut self, ctx: &mut Context<Self>, msg: DisConnected) -> Result<()> {
        let index = msg.0;
        error!("{} {} disconnected", self, self.nodes[index].addr);
        self.nodes[index].stream = None;
        ctx.address().send(Connect(index))
    }
}

pub struct SendMsgRandom<T: Message<Result = ()> + HoneybeeMessage>(pub T);
impl<T: Message<Result = ()> + HoneybeeMessage> Message for SendMsgRandom<T> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T> Handler<SendMsgRandom<T>> for TcpConnectersActor
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendMsgRandom<T>) -> Result<()> {
        let mut rng = thread_rng();
        let node = self
            .nodes
            .choose(&mut rng)
            .ok_or_else(|| anyhow!("nodes None"))?;

        if let Some(ref stream) = node.stream {
            stream.send_remote(msg.0)?;
        } else {
            error!(
                "{} send msg:{:?} to {} failed, not connected",
                self, msg.0, node
            );
        }
        Ok(())
    }
}

pub struct CallMsgRandom<T: Message<Result = R> + HoneybeeMessage, R: RpcResult>(
    pub oneshot::Sender<std::result::Result<R, RpcResponseError>>,
    pub T,
);
impl<T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message for CallMsgRandom<T, R> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T, R> Handler<CallMsgRandom<T, R>> for TcpConnectersActor
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: CallMsgRandom<T, R>) -> Result<()> {
        let mut rng = thread_rng();
        let node = self
            .nodes
            .choose(&mut rng)
            .ok_or_else(|| anyhow!("nodes None"))?;

        if let Some(ref stream) = node.stream {
            stream.send(super::tcpstream::CallMsg(msg.0, msg.1))?;
        } else {
            error!(
                "{} call msg:{:?} to {} failed, not connected",
                self, msg.1, node
            );
        }
        Ok(())
    }
}

pub struct AsyncCallMsgRandom<
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
>(pub Addr<A>, pub T);

impl<A: Actor + AnyHandler<R>, T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message
    for AsyncCallMsgRandom<A, T, R>
{
    type Result = ();
}

#[async_trait::async_trait]
impl<A, T, R> Handler<AsyncCallMsgRandom<A, T, R>> for TcpConnectersActor
where
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: AsyncCallMsgRandom<A, T, R>,
    ) -> Result<()> {
        let mut rng = thread_rng();
        let node = self
            .nodes
            .choose(&mut rng)
            .ok_or_else(|| anyhow!("nodes None"))?;

        if let Some(ref stream) = node.stream {
            let _ = stream.send(super::tcpstream::AsyncCallMsg(msg.0, msg.1));
        } else {
            error!(
                "{} async call msg:{:?} to {} failed, not connected",
                self, msg.1, node
            );
        }
        Ok(())
    }
}

pub struct SendMsgSpecific<T: Message<Result = ()> + HoneybeeMessage>(pub String, pub T);
impl<T: Message<Result = ()> + HoneybeeMessage> Message for SendMsgSpecific<T> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T> Handler<SendMsgSpecific<T>> for TcpConnectersActor
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendMsgSpecific<T>) -> Result<()> {
        let node = self
            .nodes
            .iter()
            .find(|node| node.remote_node_name == msg.0)
            .ok_or_else(|| anyhow!("nodes None"))?;

        if let Some(ref stream) = node.stream {
            stream.send_remote(msg.1)?;
        } else {
            error!(
                "{} send msg:{:?} to {} failed, not connected",
                self, msg.1, node
            );
        }
        Ok(())
    }
}

pub struct CallMsgSpecific<T: Message<Result = R> + HoneybeeMessage, R: RpcResult>(
    pub String,
    pub oneshot::Sender<std::result::Result<R, RpcResponseError>>,
    pub T,
);
impl<T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message for CallMsgSpecific<T, R> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T, R> Handler<CallMsgSpecific<T, R>> for TcpConnectersActor
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: CallMsgSpecific<T, R>) -> Result<()> {
        let node = self
            .nodes
            .iter()
            .find(|node| node.remote_node_name == msg.0)
            .ok_or_else(|| anyhow!("nodes None"))?;

        if let Some(ref stream) = node.stream {
            stream.send(super::tcpstream::CallMsg(msg.1, msg.2))?;
        } else {
            error!(
                "{} call msg:{:?} to {} failed, not connected",
                self, msg.2, node
            );
        }
        Ok(())
    }
}

pub struct AsyncCallMsgSpecific<
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
>(pub String, pub Addr<A>, pub T);

impl<A: Actor + AnyHandler<R>, T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message
    for AsyncCallMsgSpecific<A, T, R>
{
    type Result = ();
}

#[async_trait::async_trait]
impl<A, T, R> Handler<AsyncCallMsgSpecific<A, T, R>> for TcpConnectersActor
where
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: AsyncCallMsgSpecific<A, T, R>,
    ) -> Result<()> {
        let node = self
            .nodes
            .iter()
            .find(|node| node.remote_node_name == msg.0)
            .ok_or_else(|| anyhow!("nodes None"))?;

        if let Some(ref stream) = node.stream {
            let _ = stream.send(super::tcpstream::AsyncCallMsg(msg.1, msg.2));
        } else {
            error!(
                "{} async call msg:{:?} to {} failed, not connected",
                self, msg.2, node
            );
        }
        Ok(())
    }
}

pub struct SendMsgHash<T: Message<Result = ()> + HoneybeeMessage>(pub usize, pub T);
impl<T: Message<Result = ()> + HoneybeeMessage> Message for SendMsgHash<T> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T> Handler<SendMsgHash<T>> for TcpConnectersActor
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendMsgHash<T>) -> Result<()> {
        let size = self.nodes.len();
        let key = msg.0;
        let node = &self.nodes[key % size];

        if let Some(ref stream) = node.stream {
            stream.send_remote(msg.1)?;
        } else {
            error!(
                "{} send msg:{:?} to {} failed, not connected",
                self, msg.1, node
            );
        }
        Ok(())
    }
}

pub struct CallMsgHash<T: Message<Result = R> + HoneybeeMessage, R: RpcResult>(
    pub usize,
    pub oneshot::Sender<std::result::Result<R, RpcResponseError>>,
    pub T,
);
impl<T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message for CallMsgHash<T, R> {
    type Result = ();
}
#[async_trait::async_trait]
impl<T, R> Handler<CallMsgHash<T, R>> for TcpConnectersActor
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: CallMsgHash<T, R>) -> Result<()> {
        let size = self.nodes.len();
        let key = msg.0;
        let node = &self.nodes[key % size];

        if let Some(ref stream) = node.stream {
            stream.send(super::tcpstream::CallMsg(msg.1, msg.2))?;
        } else {
            error!(
                "{} call msg:{:?} to {} failed, not connected",
                self, msg.2, node
            );
        }
        Ok(())
    }
}

pub struct AsyncCallMsgHash<
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
>(pub usize, pub Addr<A>, pub T);

impl<A: Actor + AnyHandler<R>, T: Message<Result = R> + HoneybeeMessage, R: RpcResult> Message
    for AsyncCallMsgHash<A, T, R>
{
    type Result = ();
}

#[async_trait::async_trait]
impl<A, T, R> Handler<AsyncCallMsgHash<A, T, R>> for TcpConnectersActor
where
    A: Actor + AnyHandler<R>,
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(
        &mut self,
        _ctx: &mut Context<Self>,
        msg: AsyncCallMsgHash<A, T, R>,
    ) -> Result<()> {
        let size = self.nodes.len();
        let key = msg.0;
        let node = &self.nodes[key % size];

        if let Some(ref stream) = node.stream {
            let _ = stream.send(super::tcpstream::AsyncCallMsg(msg.1, msg.2));
        } else {
            error!(
                "{} async call msg:{:?} to {} failed, not connected",
                self, msg.2, node
            );
        }
        Ok(())
    }
}

pub struct SendMsgAll<T: Message<Result = ()> + HoneybeeMessage + Clone>(pub T);
impl<T: Message<Result = ()> + HoneybeeMessage + Clone> Message for SendMsgAll<T> {
    type Result = ();
}

#[async_trait::async_trait]
impl<T> Handler<SendMsgAll<T>> for TcpConnectersActor
where
    T: Message<Result = ()> + HoneybeeMessage + Clone,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendMsgAll<T>) -> Result<()> {
        for node in &self.nodes {
            if let Some(ref stream) = node.stream {
                stream.send_remote(msg.0.clone())?;
            } else {
                error!(
                    "{} send msg:{:?} to {} failed, not connected",
                    self, msg.0, node
                );
            }
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl RemotesAddrExt for Addr<TcpConnectersActor> {
    fn send_remote_random<T>(&self, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        self.send(SendMsgRandom(msg))
    }

    async fn call_remote_random<T, R>(
        &self,
        msg: T,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let (tx, rx) = oneshot::channel();

        self.send(CallMsgRandom(tx, msg))
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?;

        rx.await
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?
    }

    fn async_call_remote_random<A, T, R>(&self, source: Addr<A>, msg: T) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        self.send(AsyncCallMsgRandom(source, msg))
    }

    fn send_remote_specific<T>(&self, remote_node_name: String, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        self.send(SendMsgSpecific(remote_node_name, msg))
    }

    async fn call_remote_specific<T, R>(
        &self,
        remote_node_name: String,
        msg: T,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let (tx, rx) = oneshot::channel();

        self.send(CallMsgSpecific(remote_node_name, tx, msg))
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?;

        rx.await
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?
    }

    fn async_call_remote_specific<A, T, R>(
        &self,
        remote_node_name: String,
        source: Addr<A>,
        msg: T,
    ) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        self.send(AsyncCallMsgSpecific(remote_node_name, source, msg))
    }

    fn send_remote_hash<T>(&self, key: usize, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        self.send(SendMsgHash(key, msg))
    }

    async fn call_remote_hash<T, R>(
        &self,
        key: usize,
        msg: T,
    ) -> std::result::Result<T::Result, RpcResponseError>
    where
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        let (tx, rx) = oneshot::channel();

        self.send(CallMsgHash(key, tx, msg))
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?;

        rx.await
            .map_err(|err| RpcResponseError::RpcError(err.to_string()))?
    }

    fn async_call_remote_hash<A, T, R>(&self, key: usize, source: Addr<A>, msg: T) -> Result<()>
    where
        A: Actor + AnyHandler<R>,
        T: Message<Result = R> + HoneybeeMessage,
        R: RpcResult,
    {
        self.send(AsyncCallMsgHash(key, source, msg))
    }

    fn send_remote_all<T>(&self, msg: T) -> Result<()>
    where
        T: Message<Result = ()> + HoneybeeMessage + Clone,
    {
        self.send(SendMsgAll(msg))
    }
}

impl UserSender for Addr<TcpConnectersActor> {
    fn send_to_user<T>(&self, uid: i64, msg: T)
    where
        T: 'static + prost::Message + MessageId,
    {
        let bytes = {
            let mut buf = Vec::with_capacity(msg.encoded_len());
            if let Err(err) = msg.encode(&mut buf) {
                error!("send to user uid:{} msg:{} err:{}", uid, T::ID, err);
                return;
            }
            buf
        };
        if let Err(err) = self.send_remote_hash(
            uid as _,
            UserMessageForward {
                uid,
                msg_id: T::ID,
                bytes,
            },
        ) {
            error!("send to user uid:{} msg:{} err:{}", uid, T::ID, err);
        }
    }
}

impl AllUserSender for Addr<TcpConnectersActor> {
    fn send_to_all_user<T>(&self, msg: T)
    where
        T: 'static + prost::Message + MessageId,
    {
        let bytes = {
            let mut buf = Vec::with_capacity(msg.encoded_len());
            if let Err(err) = msg.encode(&mut buf) {
                error!("send msg to all user:{} err:{}", T::ID, err);
                return;
            }
            buf
        };
        if let Err(err) = self.send_remote_random(AllUserMessageForward {
            msg_id: T::ID,
            bytes,
        }) {
            error!("send msg to all user:{} err:{}", T::ID, err);
        }
    }
}

impl AppUserSender for Addr<TcpConnectersActor> {
    fn send_to_app_user<T>(&self, uid: i64, msg: T)
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        match bincode::serialize(&msg) {
            Ok(bytes) => {
                if let Err(err) = self.send_remote_hash(
                    uid as _,
                    UserMessageApp {
                        uid,
                        msg_id: T::ID,
                        bytes,
                    },
                ) {
                    error!("send to app user uid:{} msg:{:?} failed:{}", uid, msg, err);
                }
            }
            Err(err) => error!("send to app user uid:{} msg:{:?} failed:{}", uid, msg, err),
        }
    }
}

impl AllAppUserSender for Addr<TcpConnectersActor> {
    fn send_to_all_app_user<T>(&self, msg: T)
    where
        T: Message<Result = ()> + HoneybeeMessage,
    {
        match bincode::serialize(&msg) {
            Ok(bytes) => {
                if let Err(err) = self.send_remote_random(AllUserMessageApp {
                    msg_id: T::ID,
                    bytes,
                }) {
                    error!("send to all app user msg:{:?} failed:{}", msg, err);
                }
            }
            Err(err) => error!("send to all app user msg:{:?} failed:{}", msg, err),
        }
    }
}

impl UmsgSender for Addr<TcpConnectersActor> {
    fn reliable_send_to_user<T>(&self, uid: i64, msg_type: umsg_db::UmsgType, msg: T)
    where
        T: Serialize + Debug,
    {
        let msg_doc = match mongodb::bson::to_document(&msg) {
            Ok(msg_doc) => msg_doc,
            Err(err) => {
                error!(
                    "uid:{} reliable send umsg type:{:?} msg:{:?} to_document err:{}",
                    uid, msg_type, msg, err
                );
                return;
            }
        };
        let mut msg_bytes = Vec::new();
        if let Err(err) = msg_doc.to_writer(&mut msg_bytes) {
            error!(
                "uid:{} reliable send umsg type:{:?} msg:{:?} doc to writer err:{}",
                uid, msg_type, msg, err
            );
            return;
        }
        let umsg = UmsgForward {
            uid,
            msg_type,
            msg_bytes,
        };
        if let Err(err) = self.send_remote_hash(uid as _, umsg) {
            error!(
                "uid:{} reliable send umsg type:{:?} msg:{:?} err:{}",
                uid, msg_type, msg, err
            );
        }
    }
}

impl GmsgSender for Addr<TcpConnectersActor> {
    fn reliable_send_to_all_user<T>(&self, msg_type: gmsg_db::GmsgType, msg: T, expire: Option<i64>)
    where
        T: Serialize + Debug,
    {
        let msg_doc = match mongodb::bson::to_document(&msg) {
            Ok(msg_doc) => msg_doc,
            Err(err) => {
                error!(
                    "reliable send gmsg type:{:?} msg:{:?} to_document err:{}",
                    msg_type, msg, err
                );
                return;
            }
        };
        let mut msg_bytes = Vec::new();
        if let Err(err) = msg_doc.to_writer(&mut msg_bytes) {
            error!(
                "reliable send gmsg type:{:?} msg:{:?} doc to writer err:{}",
                msg_type, msg, err
            );
            return;
        }
        let gmsg = GmsgForward {
            msg_type,
            msg_bytes,
            expire,
        };
        if let Err(err) = self.send_remote_random(gmsg) {
            error!(
                "reliable send gmsg type:{:?} msg:{:?} err:{}",
                msg_type, msg, err
            );
        }
    }
}
